package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.bw.gmall.realtime.common.util.SQLUtil.getKafkaSourceSQL;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dwd.db.app
 * @className: DwdTradeOrderCancelDetail
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/26 10:33
 * @version: 1.0
 */

public class DwdTradeOrderPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS, 4, 10016);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        // 1. 读取下单事务事实表
        extracted(tableEnv, groupId);
//        tableEnv.sqlQuery("select * from dwd_trade_order_detail").execute().print();

        //读取topic_db的数据
        readOdsDb(tableEnv, Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);
        //读取dim层的数据
        readBaseDic(tableEnv);

        //过滤出payment_info数据
        Table table1 = getPayment_info(tableEnv);
        tableEnv.createTemporaryView("payment_info", table1);
//        table1.execute().print();
        // 5、Join操作
        Table joinTable = getJoin(tableEnv);
        tableEnv.createTemporaryView("joinTable",joinTable);
        // 关联字典表
//        joinTable.execute().print();
        Table table = getTable(tableEnv);
//        table.execute().print();

        //写入kafka
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS +"(\n" +
                "    id string,\n" +
                "    order_id string,\n" +
                "    user_id string,\n" +
                "    sku_id string,\n" +
                "    sku_name string,\n" +
                "    province_id string,\n" +
                "    activity_id string,\n" +
                "    coupon_id string,\n" +
                "    date_id string,\n" +
                "    create_time string,\n" +
                "    sku_num string,\n" +
                "    split_original_amount DECIMAL(10, 2),\n" +
                "    split_activity_amount DECIMAL(10, 2),\n" +
                "    split_coupon_amount DECIMAL(10, 2),\n" +
                "    split_total_amount DECIMAL(10, 2),\n" +
                "    payment_type string,\n" +
                "    payment_type_name string,\n" +
                "    callback_time string,\n" +
                "    ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));
        table.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);



    }

    private static void extracted(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql(
                "create table dwd_trade_order_detail(\n" +
                        "  id String,\n" +
                        "  order_id String,\n" +
                        "  user_id String,\n" +
                        "  sku_id String,\n" +
                        "  sku_name String,\n" +
                        "  province_id String,\n" +
                        "  activity_id String,\n" +
                        "  coupon_id String,\n" +
                        "  date_id String,\n" +
                        "  create_time String,\n" +
                        "  sku_num String,\n" +
                        "  split_original_amount DECIMAL(10, 2),\n" +
                        "  split_activity_amount DECIMAL(10, 2),\n" +
                        "  split_coupon_amount DECIMAL(10, 2),\n" +
                        "  split_total_amount DECIMAL(10, 2),\n" +
                        "  ts bigint,\n" +
                        "  row_time  as TO_TIMESTAMP_LTZ(ts, 0),\n" +
                        "  WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND\n" +
                        ")" + getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, groupId));
    }

    private static Table getTable(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select id,\n" +
                "order_id,\n" +
                "user_id,\n" +
                "sku_id,\n" +
                "sku_name,\n" +
                "province_id,\n" +
                "activity_id,\n" +
                "coupon_id,\n" +
                "date_id,\n" +
                "create_time,\n" +
                "sku_num,\n" +
                "split_original_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "split_total_amount,\n" +
                "payment_type,\n" +
                "info.dic_name as payment_type_name,\n" +
                "callback_time,\n" +
                "ts\n" +
                " from joinTable as j \n" +
                " join base_dic FOR SYSTEM_TIME AS OF j.proc_time AS c\n" +
                "    ON j.payment_type = c.rowkey;");
    }

    private static Table getJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery(" select d.id ,\n" +
                " d.order_id,\n" +
                " d.user_id,\n" +
                " d.sku_id,\n" +
                " d.sku_name,\n" +
                " d.province_id,\n" +
                " d.activity_id,\n" +
                " d.coupon_id,\n" +
                " d.date_id,\n" +
                " d.create_time,\n" +
                " d.sku_num,\n" +
                " d.split_original_amount,\n" +
                " d.split_activity_amount,\n" +
                " d.split_coupon_amount,\n" +
                " d.split_total_amount,\n" +
                " p.payment_type,\n" +
                " p.callback_time,\n" +
                " p.ts,\n" +
                " p.row_time,\n" +
                " p.proc_time\n" +
                " from payment_info p ,dwd_trade_order_detail d \n" +
                " where p.order_id = d.order_id \n" +
                " AND p.row_time BETWEEN d.row_time - INTERVAL '15' MINUTE AND p.row_time + INTERVAL '5' SECOND");
    }

    private static Table getPayment_info(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("SELECT \n" +
                "    `data`['id'] AS id,\n" +
                "    `data`['order_id'] AS order_id,\n" +
                "    `data`['user_id'] AS user_id,\n" +
                "    `data`['payment_type'] AS payment_type,\n" +
                "    `data`['callback_time'] AS callback_time,\n" +
                "    `ts`,\n" +
                "    `row_time`,\n" +
                "    proc_time AS proc_time \n" +
                "FROM topic_db\n" +
                "WHERE `database` = 'gmall'\n" +
                "  AND `table` = 'payment_info'\n" +
                "  AND `type` = 'update'\n" +
                "  AND `old`['payment_status'] IS NOT NULL\n" +
                "  AND `data`['payment_status'] = '1602'");
    }
}